package pubM

import (
	"db2s/Meta"
	"db2s/arg"
	"db2s/conn"
	"db2s/global"
	"db2s/parDef"
	"db2s/ref"
	mq "db2s/topic-mq"
	"sort"
	"sync"
	"sync/atomic"
)

const (
	ChannelDefault = 10000000
)

func tableObjectSeqUnique(TableObject []*arg.InputTableAttributes) (p []int) {
	var m = make(map[int]int)
	for _, v := range TableObject {
		if _, ok := m[v.GroupSeq]; !ok {
			m[v.GroupSeq]++
			p = append(p, v.GroupSeq)
		}
	}
	sort.Ints(p)
	return
}
func tableObjectSeqUniqueSort(TableObject []*arg.InputTableAttributes) (result []any) {
	for _, s := range tableObjectSeqUnique(TableObject) {
		var r = make(chan any, ChannelDefault)
		for _, q := range TableObject {
			if s == q.GroupSeq {
				r <- q
			}
		}
		close(r)
		result = append(result, r)
	}
	return
}

type ObjectFlow struct {
	Mtc            int
	Drive          []string
	DriveJoin      string
	Source, Target conn.ConEr
	IgnoreSum      *int64
	Object         []*arg.InputTableAttributes
	Flow           chan any
}

func getTableObject(flow ObjectFlow, v1 *arg.InputTableAttributes) (err error) {
	var exist bool
	r := Meta.NewMeta(flow.Drive, flow.DriveJoin, v1.TableAttributes.Role)
	r.Parameter1 = append(r.Parameter1, parDef.Parameter{Con: flow.Source.GetDB("single"), Object: parDef.Object{Schema: v1.Schema, Table: v1.Table}})
	r.Parameter1 = append(r.Parameter1, parDef.Parameter{Con: flow.Target.GetDB("single"), Object: parDef.Object{Schema: v1.MSchema, Table: v1.MTable}})
	if exist, err = r.GetObjectExistence(); err != nil || exist {
		return
	}
	atomic.SwapInt64(v1.TableStatus, -1)
	return
}
func objectExistPointQuit(event string, _ any) (r any) {
	ref.MyWaitDel("objectExist")
	return
}
func objectExistPointInit(event string, f any) (r any) {
	var err error
	if f == nil || len(f.([]any)) == 0 {
		return
	}
	_, chanParameter, userParameter := schemaMetaExecFuncParameterSplit(f.([]any))
	re := chanParameter.(*arg.InputTableAttributes)
	u := userParameter[0].(ObjectFlow)
	if err = getTableObject(u, re); err != nil {
		err = ref.ErrAddPrintf("objectExistPointInit", err)
	}
	u.Flow <- *re
	return
}
func ignorePointContinue(event string, f any) (r any) {
	if f == nil || len(f.([]any)) == 0 {
		return
	}
	_, chanParameter, userParameter := schemaMetaExecFuncParameterSplit(f.([]any))
	v := chanParameter.(*arg.InputTableAttributes)
	u := userParameter[0].(ObjectFlow)
	if v.Ignore {
		atomic.AddInt64(u.IgnoreSum, 1)
		return true
	}
	return false
}
func objectExistInit(flow ObjectFlow, m any) mq.MonitorMsgMq {
	var wg sync.WaitGroup
	return mq.NewMonitorMsgMq1().SetSendMsg(m.(chan any)).SetCurryLimit(mq.SendCurryLimit{
		Wg:       &wg,
		Switch:   true,
		CurrySum: make(chan struct{}, flow.Mtc),
	}).SetNormalQuitFunc(global.EFunc{
		FuncName: objectExistPointQuit}).SetAbnormalQuitFunc(global.EFunc{
		FuncName: objectExistPointQuit}).SetContinueFunc(global.EFunc{
		FuncName: ignorePointContinue, Params: flow}).SetExecFunc([]global.EFunc{{FuncName: objectExistPointInit, Params: flow}})

}

func NewObjectFlow() (flow ObjectFlow) {
	return
}
func ObjectExist(flow ObjectFlow) (err error) {
	var event = "[NewObjectFlow]"
	for _, tg := range tableObjectSeqUniqueSort(flow.Object) {
		ref.MyWaitAdd("objectExist")
		if err = objectExistInit(flow, tg).NormalChanSendMsg(event); err != nil {
			return
		}
	}
	return
}
